package service

import (
	"census/common/utils"
	"census/context"
	"fmt"
	"github.com/golib/logger"
	"github.com/golib/rabbitmq"
	"sync/atomic"
	"time"
)

type MQService struct {
	CommonService
	queueName string
	appName   string
	sid       string
	count     int64
	ch        chan *context.AppRequest
}

func NewMQService(sid, queueName, appName string) *MQService {
	return &MQService{
		sid:       sid,
		queueName: queueName,
		appName:   appName,
		ch:        make(chan *context.AppRequest),
	}
}

func (mq *MQService) SID() string {
	return mq.sid
}

func (mq *MQService) Reload() (err error) {
	go mq.Run()
	return
}

func (mq *MQService) Shutdown() (err error) {
	t := time.Tick(100 * time.Millisecond)
	for {
		select {
		case <-t:
			if atomic.LoadInt64(&mq.count) == 0 {
				return
			}
		}
	}
}

func (mq *MQService) Run() (err error) {
	conn := rabbitmq.DefaultInstance()
	c, err := conn.Consumer(mq.queueName)
	if err != nil {
		fmt.Printf("[rabbitMQ connect error]: appName:(%v) queueName:(%v)", mq.appName, mq.queueName)
		logger.Warning("mq-conn", utils.MakeUUID(), fmt.Sprintf("[rabbitMQ connect error]: appName:(%v) queueName:(%v)", mq.appName, mq.queueName))
		return err
	}
	for {
		//get mq body
		mqMessage := c.Fetch()
		//transaction begin 1
		atomic.AddInt64(&mq.count, 1)
		var request = new(context.AppRequest)
		var err = request.Decode(mqMessage.Body)
		if err != nil {
			logger.Warning("mq-request", utils.MakeUUID(), "decode error.please check!!", "appName:["+mq.appName+"]", "queueName:["+mq.queueName+"]", "genTime:", request.GenTime, "raw data", string(mqMessage.Body))
			continue
		}
		err = mq.Send(request)
		if err != nil {
			//transaction fail 0
			atomic.AddInt64(&mq.count, -1)
			switch err {
			case errServiceSignalStop:
				c.Cancel()
				logger.Info("mq-request", utils.MakeUUID(), "mq service receive stop sigal", "appName:["+mq.appName+"]", "queueName:["+mq.queueName+"]", "error info", err.Error())
				return err
			default:
				logger.Warning("mq-request", utils.MakeUUID(), "mq send msg error", "appName:["+mq.appName+"]", "queueName:["+mq.queueName+"]", "error info", err.Error())
				continue
			}
		}
		//mq ack
		mqMessage.Ack(false)
		logger.Info("mq-request", utils.MakeUUID(), "appName:["+mq.appName+"]", "queueName:["+mq.queueName+"]", "genTime:", "genTime:", request.GenTime, "raw data", string(mqMessage.Body))
		//transaction success 0
		atomic.AddInt64(&mq.count, -1)
	}
	return
}
